package com.skycan.plc.adapter.mqtt;

import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
import com.skycan.plc.adapter.config.MqttConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * MQTT客户端
 */
public class MqttClient {
    private static final Logger logger = LoggerFactory.getLogger(MqttClient.class);
    private static final MqttClient INSTANCE = new MqttClient();

    private Mqtt5BlockingClient client;
    private final MqttConfig mqttConfig;
    private boolean connected = false;

    private MqttClient() {
        mqttConfig = MqttConfig.getInstance();
        init();
    }

    public static MqttClient getInstance() {
        return INSTANCE;
    }

    /**
     * 初始化MQTT客户端
     */
    private void init() {
        String clientId = mqttConfig.getClientId() + "-" + UUID.randomUUID().toString().substring(0, 8);

        client = Mqtt5Client.builder()
                .identifier(clientId)
                .serverHost(extractHost(mqttConfig.getBrokerUrl()))
                .serverPort(extractPort(mqttConfig.getBrokerUrl()))
                .automaticReconnectWithDefaultConfig()
                .buildBlocking();

        logger.info("MQTT客户端初始化完成: {}", clientId);
    }

    /**
     * 连接MQTT服务器
     */
    public boolean connect() {
        if (connected) {
            return true;
        }

        try {
            client.connectWith()
                    .cleanStart(true)
                    .keepAlive(60)
                    .simpleAuth()
                    .username(mqttConfig.getUsername())
                    .password(mqttConfig.getPassword().getBytes(StandardCharsets.UTF_8))
                    .applySimpleAuth()
                    .send();

            connected = true;
            logger.info("MQTT服务器连接成功: {}", mqttConfig.getBrokerUrl());
            return true;
        } catch (Exception e) {
            logger.error("MQTT服务器连接失败: {}, 错误: {}", mqttConfig.getBrokerUrl(), e.getMessage());
            return false;
        }
    }

    /**
     * 断开MQTT连接
     */
    public void disconnect() {
        if (connected && client != null) {
            try {
                client.disconnect();
                connected = false;
                logger.info("MQTT服务器断开连接");
            } catch (Exception e) {
                logger.error("MQTT服务器断开连接失败: {}", e.getMessage());
            }
        }
    }

    /**
     * 发布消息
     *
     * @param topic   主题
     * @param payload 消息内容
     * @return 是否发布成功
     */
    public boolean publish(String topic, String payload) {
        if (!connected && !connect()) {
            return false;
        }

        try {
            client.publishWith()
                    .topic(topic)
                    .qos(MqttQos.AT_LEAST_ONCE)
                    .payload(payload.getBytes(StandardCharsets.UTF_8))
                    .send();

            logger.debug("MQTT消息发布成功: 主题={}, 内容长度={}", topic, payload.length());
            return true;
        } catch (Exception e) {
            logger.error("MQTT消息发布失败: 主题={}, 错误={}", topic, e.getMessage());
            return false;
        }
    }

    /**
     * 订阅主题
     *
     * @param topic          主题
     * @param messageHandler 消息处理器
     * @return 是否订阅成功
     */
    public boolean subscribe(String topic, Consumer<String> messageHandler) {
        if (!connected && !connect()) {
            return false;
        }

        try {
            client.toAsync().subscribeWith()
                    .topicFilter(topic)
                    .qos(MqttQos.AT_LEAST_ONCE)
                    .callback(publish -> {
                        String payload = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
                        logger.debug("收到MQTT消息: 主题={}, 内容长度={}", publish.getTopic(), payload.length());

                        try {
                            messageHandler.accept(payload);
                        } catch (Exception e) {
                            logger.error("处理MQTT消息异常: {}", e.getMessage());
                        }
                    })
                    .send()
                    .get(5, TimeUnit.SECONDS);

            logger.info("MQTT主题订阅成功: {}", topic);
            return true;
        } catch (Exception e) {
            logger.error("MQTT主题订阅失败: {}, 错误={}", topic, e.getMessage());
            return false;
        }
    }

    /**
     * 取消订阅
     *
     * @param topic 主题
     * @return 是否取消成功
     */
    public boolean unsubscribe(String topic) {
        if (!connected) {
            return true;
        }

        try {
            client.unsubscribeWith()
                    .topicFilter(topic)
                    .send();

            logger.info("MQTT主题取消订阅成功: {}", topic);
            return true;
        } catch (Exception e) {
            logger.error("MQTT主题取消订阅失败: {}, 错误={}", topic, e.getMessage());
            return false;
        }
    }

    /**
     * 从URL中提取主机名
     */
    private String extractHost(String url) {
        if (url.contains("://")) {
            url = url.split("://")[1];
        }

        if (url.contains(":")) {
            return url.split(":")[0];
        }

        return url;
    }

    /**
     * 从URL中提取端口
     */
    private int extractPort(String url) {
        if (url.contains("://")) {
            url = url.split("://")[1];
        }

        if (url.contains(":")) {
            try {
                return Integer.parseInt(url.split(":")[1]);
            } catch (NumberFormatException e) {
                // 默认使用MQTT标准端口
                return 1883;
            }
        }

        // 默认使用MQTT标准端口
        return 1883;
    }

    /**
     * 检查是否已连接
     */
    public boolean isConnected() {
        return connected;
    }
}